package com.sam.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author: wm.xue
 * @date: 2021/11/3 18:24
 * @description:
 */
public class _01_MyProducer {
    public static void main(String[] args) {
        // KafkaProducer配置
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.100.17.108:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        configs.put(ProducerConfig.RETRIES_CONFIG, 2);
        configs.put(ProducerConfig.ACKS_CONFIG, "1");
//        configs.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configs);
        for (int i = 0; i < 50; i++) {
            ProducerRecord record = new ProducerRecord("tp_02_demo", 0, "key：" + i, "value: " + i);

            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println("消息的主题：" + metadata.topic());
                    System.out.println("消息的分区号：" + metadata.partition());
                    System.out.println("消息的偏移量：" + metadata.offset());
                }
            });
        }
        producer.close();
    }
}
